package com.ayu.shortlink.project.mq.consumer;

import com.ayu.shortlink.project.common.constant.RocketMQConstant;
import com.ayu.shortlink.project.common.constant.ShortLinkConstant;
import com.ayu.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.ayu.shortlink.project.mq.basics.base.MessageWrapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

/**
 * 数据聚合消费者
 */
@RocketMQMessageListener(
        consumerGroup = RocketMQConstant.UNION_DATA_CONSUMER,
        topic = RocketMQConstant.UNION_DATA_TOPIC
)
@Component
@Slf4j
@RequiredArgsConstructor
public class ShortLinkStatsUnionSaveConsumer implements RocketMQListener<MessageWrapper<ShortLinkStatsRecordDTO>> {
    private final StringRedisTemplate stringRedisTemplate;

    /**
     * 处理消息聚合
     * @param shortLinkStatsRecordDTOMessageWrapper
     */
    @Override
    public void onMessage(MessageWrapper<ShortLinkStatsRecordDTO> shortLinkStatsRecordDTOMessageWrapper) {
        ShortLinkStatsRecordDTO message = shortLinkStatsRecordDTOMessageWrapper.getMessage();

        // 进行聚合操作 格式 前缀_短URL_原URL
        String key = String.format(ShortLinkConstant.PV_UNION_KEY, message.getFullShortUrl(), message.getOriginUrl());

        // 执行原子递增
        stringRedisTemplate.opsForValue().increment(key);
        // 放入SET
        String var = message.getFullShortUrl() + "_" + message.getOriginUrl();
        stringRedisTemplate.opsForSet().add(ShortLinkConstant.UNION_SET_KEY,var);
    }
}
